package com.miaozhen.datainsight.web.config;

import com.miaozhen.datainsight.web.jobMessaging.JobStateListener;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.boot.autoconfigure.kafka.ConcurrentKafkaListenerContainerFactoryConfigurer;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.listener.KafkaMessageListenerContainer;
import org.springframework.kafka.listener.config.ContainerProperties;

import java.util.Map;

/**
 * 使用消息队列和后端计算服务通信 之 kafka配置
 *
 * @author tzp
 * @date 2017/11/13
 */
@Configuration
@EnableKafka
@Slf4j
public class KafkaConfig {
    /**
     * Customized ProducerFactory bean.
     *
     * @param properties the kafka properties.
     * @return the bean.
     */
    @Bean
    public ProducerFactory kafkaProducerFactory(KafkaProperties properties) {
        Map<String, Object> producerProperties = properties.buildProducerProperties();

        return new DefaultKafkaProducerFactory<String, String>(producerProperties);
    }

    /**
     * Customized ConsumerFactory bean. essential
     *
     * @param properties the kafka properties.
     * @return the bean.
     */
    @Bean
    public ConsumerFactory<Object, Object> kafkaConsumerFactory(KafkaProperties properties) {
        Map<String, Object> consumerProperties = properties.buildConsumerProperties();

        return new DefaultKafkaConsumerFactory<>(consumerProperties);
    }

    /**
     * listener container factory
     */
    @Bean
    public ConcurrentKafkaListenerContainerFactory<Object, Object> kafkaListenerContainerFactory(
            ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
            ConsumerFactory<Object, Object> cf) {
        ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
        configurer.configure(factory, cf);
        return factory;
    }

    /**
     * 创建container方法1. 手动用cf和containerProps创建.
     * listener container. essential
     */
    @Bean
    public KafkaMessageListenerContainer<Object, Object> registerJobStateListener(
            ConsumerFactory<Object, Object> cf,
            JobStateListener jobStateListener) {
        ContainerProperties containerProps = new ContainerProperties("seg1");
        containerProps.setMessageListener(jobStateListener);
        return new KafkaMessageListenerContainer<>(cf, containerProps);
    }

    @KafkaListener(topics = "seg1")
    public void temeListener(ConsumerRecord<Object, Object> record) {
        log.info("myreceive1" + record.key() + " " + record.value());
    }
}
